package nats

import (
	"charging_pile/netty/internal/svc"
	"charging_pile/util/protocol/mq"
	"encoding/json"
	"fmt"

	"github.com/nats-io/nats.go"
	"github.com/zeromicro/go-zero/core/logx"
	"github.com/zeromicro/go-zero/core/service"
)

type natsClient struct {
	conn   *nats.Conn
	SvcCtx *svc.ServiceContext
}

// HandNettyMsg implements def.HandNettyMsg.
func (nc *natsClient) HandNettyMsg(msg *mq.Msg) {
	b, err := json.Marshal(msg)
	if err != nil {
		return
	}
	nc.Publish("server", b)
}

type TopicHandle func(*natsClient, []byte)

func MustNewNatsConn(svcCtx *svc.ServiceContext) service.Service {
	//输入nats服务的地址和端口
	localnc := &natsClient{
		// conn:   natsConn,
		SvcCtx: svcCtx,
	}
	addr := svcCtx.Config.MqAddr
	natsConn, err := nats.Connect(addr, nats.ReconnectHandler(func(nc *nats.Conn) {
		localnc.conn = nc
		localnc.MqSubscribeTopic(localnc.SvcCtx.LocalTopic, GroupTopicHandler)
		localnc.MqSubscribeGroupTopic("iot", "device", GroupTopicHandler)
	}))
	if err != nil {
		panic(err)
	}
	localnc.conn = natsConn
	svcCtx.MQ = localnc

	return localnc
}

// 订阅消息，group 只有一个收到
func (nc *natsClient) MqSubscribeGroupTopic(topic, group string, cb TopicHandle) {
	nc.conn.QueueSubscribe(topic, group, func(msgData *nats.Msg) {
		logx.Infof("[mq] groupTopic:%s-%s:%+v", group, topic, msgData)
		cb(nc, msgData.Data)
	})
}

// 订阅消息，只要订阅都能收到
func (nc *natsClient) MqSubscribeTopic(topic string, cb TopicHandle) {
	nc.conn.Subscribe(topic, func(msgData *nats.Msg) {
		// logx.Infof("[mq] topic:%s:%+v", topic, msgData)
		cb(nc, msgData.Data)
	})
}

// 发布消息
func (nc *natsClient) Publish(topic string, payload []byte) {
	nc.conn.Publish(topic, payload)

}

func (nc *natsClient) Start() {
	topic := nc.SvcCtx.LocalTopic
	fmt.Println("订阅消息：", topic)
	nc.MqSubscribeTopic(topic, TopicHandler)
	fmt.Println("订阅组消息：", "iot", "device")
	nc.MqSubscribeGroupTopic("iot", "device", GroupTopicHandler)
}

func (nc *natsClient) Stop() {
	nc.conn.Close()
}
